package app.dws;

import app.BaseAppV1;
import beans.TradeProvinceOrderWindow;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import common.Constant;
import function.DimAsyncFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import utils.AtguiguUtil;
import utils.DimUtil;
import utils.FlinkSinkUtil;
import utils.JdbcUtil;

import java.math.BigDecimal;
import java.sql.Connection;
import java.time.Duration;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class Dws_0402_DwsTradeProvinceOrderWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_0402_DwsTradeProvinceOrderWindow().init(
                3402,2,
                "Dws_0402_DwsTradeProvinceOrderWindow",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 2. 解析成一个 pojo
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream = parseToPojo(stream);
        // 3. 开窗聚和
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStreamWithoutDim = windowAndAgg(beanStream);
        // 4. 补充维度: 旁路缓存优化和异步优化
        //SingleOutputStreamOperator<TradeProvinceOrderWindow> resultStream = joinDim(beanStreamWithoutDim);

        addDims(beanStreamWithoutDim);

        // 5. 写出到 clickhouse 中
        //writeToClickHouse(resultStream);
    }

//    private void writeToClickHouse(SingleOutputStreamOperator<TradeProvinceOrderWindow> resultStream) {
//        resultStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_trade_province_order_window", TradeProvinceOrderWindow.class));
//    }

    private void addDims(SingleOutputStreamOperator<TradeProvinceOrderWindow> stream) {
        // 补充维度信息:
        stream
                .map(new RichMapFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow>() {

                    private Connection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn = JdbcUtil.getPhoenixConnection();
                    }

                    @Override
                    public TradeProvinceOrderWindow map(TradeProvinceOrderWindow bean) throws Exception {
                        // 根据一个 id 去对应的维度表中查询
                        // 1. sku_info
                        JSONObject baseProvince = DimUtil.readDimFromPhoenix(conn, "DIM_BASE_PROVINCE", bean.getProvinceId());
                        bean.setProvinceName(baseProvince.getString("NAME"));

                        return bean;
                    }
                })
               .addSink(FlinkSinkUtil.getClickHouseSink("dws_trade_province_order_window", TradeProvinceOrderWindow.class));
    }

//    private SingleOutputStreamOperator<TradeProvinceOrderWindow> joinDim(SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStreamWithoutDim) {
//        return  AsyncDataStream.unorderedWait(
//                beanStreamWithoutDim,
//                new DimAsyncFunction<TradeProvinceOrderWindow>(){
//                    @Override
//                    public String table() {
//                        return "DIM_BASE_PROVINCE";
//                    }
//
//                    @Override
//                    public String id(TradeProvinceOrderWindow input) {
//                        return input.getProvinceId();
//                    }
//
//                    @Override
//                    public void addDim(JSONObject dim, TradeProvinceOrderWindow input) {
//                        input.setProvinceName(dim.getString("NAME"));
//                    }
//                },
//                120,
//                TimeUnit.SECONDS
//        );
//
//    }

    private SingleOutputStreamOperator<TradeProvinceOrderWindow> windowAndAgg(SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.
                                <TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((bean,ts)->bean.getTs()))

                .keyBy(TradeProvinceOrderWindow::getProvinceId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeProvinceOrderWindow>() {
                            @Override
                            public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow value1, TradeProvinceOrderWindow value2) throws Exception {
                                value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                                value1.getUserIdSet().addAll(value2.getUserIdSet());
                                value1.setFinalAmount(value1.getFinalAmount().add(value2.getFinalAmount()));
                                return value1;
                            }
                        },
                        new ProcessWindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                            @Override
                            public void process(String s,
                                                Context context,
                                                Iterable<TradeProvinceOrderWindow> elements,
                                                Collector<TradeProvinceOrderWindow> out) throws Exception {
                                TradeProvinceOrderWindow bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDatTime(context.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDatTime(context.window().getEnd()));

                                bean.setOrderCount((long) bean.getOrderIdSet().size());
                                bean.setUserCount((long) bean.getUserIdSet().size());

                                bean.setTs(System.currentTimeMillis());

                                out.collect(bean);
                            }
                        }
                )
                ;
    }

    private SingleOutputStreamOperator<TradeProvinceOrderWindow> parseToPojo(DataStreamSource<String> stream) {
        return stream
                .map(JSON::parseObject)
                .map(new MapFunction<JSONObject, TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow map(JSONObject value) throws Exception {
                        String provinceId = value.getString("province_id");
                        HashSet<String> orderIdSet = new HashSet<>();
                        orderIdSet.add(value.getString("order_id"));

                        HashSet<String> userIdSet = new HashSet<>();
                        userIdSet.add(value.getString("user_id"));

                        BigDecimal finalAmount = value.getBigDecimal("final_amount")==null ? new BigDecimal(0) : value.getBigDecimal("final_amount");
                        long ts = value.getLong("ts") * 1000;
                        return new TradeProvinceOrderWindow("","",
                                provinceId,"",orderIdSet,0L,
                                userIdSet,0L,finalAmount,ts);
                    }
                });
    }


}
